package org.groupg.practice;

import io.reactivex.rxjava3.core.BackpressureOverflowStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.CompositeDisposable;
import io.reactivex.rxjava3.schedulers.Schedulers;
import io.reactivex.rxjava3.subscribers.ResourceSubscriber;

import java.util.concurrent.TimeUnit;

public class RxJavaThreadingAndBackpressure {

    private static final CompositeDisposable disposables = new CompositeDisposable();

    public static void main(String[] args) {
        System.out.println("===== RxJava 线程调度与背压处理示例 =====");
        System.out.println("主线程: " + Thread.currentThread().getName());

        // 1. 线程调度基础
        basicThreading();

        // 2. 多级线程切换
        multiLevelThreadSwitching();

        // 3. 背压基础示例
        basicBackpressure();

        // 4. 高级背压策略
        advancedBackpressureStrategies();

        // 5. 背压与并行处理
        backpressureWithParallelProcessing();

        // 等待异步操作完成
        try {
            Thread.sleep(8000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        disposables.dispose();
    }

    // 1. 线程调度基础
    private static void basicThreading() {
        System.out.println("\n===== 1. 线程调度基础 =====");

        // 创建Observable（默认在当前线程）
        Observable<String> observable = Observable.create(emitter -> {
            System.out.println("发射线程: " + Thread.currentThread().getName());
            emitter.onNext("数据1");
            emitter.onNext("数据2");
            emitter.onNext("数据3");
            emitter.onComplete();
        });

        // 情况1：不指定线程
        observable.subscribe(item ->
                System.out.println("情况1-接收线程: " + Thread.currentThread().getName() + ", 数据: " + item));

        // 情况2：只指定subscribeOn
        observable
                .subscribeOn(Schedulers.io())
                .subscribe(item ->
                        System.out.println("情况2-接收线程: " + Thread.currentThread().getName() + ", 数据: " + item));

        // 情况3：只指定observeOn
        observable
                .observeOn(Schedulers.computation())
                .subscribe(item ->
                        System.out.println("情况3-接收线程: " + Thread.currentThread().getName() + ", 数据: " + item));

        // 情况4：同时指定subscribeOn和observeOn
        observable
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.computation())
                .subscribe(item ->
                        System.out.println("情况4-接收线程: " + Thread.currentThread().getName() + ", 数据: " + item));
    }

    // 2. 多级线程切换
    private static void multiLevelThreadSwitching() {
        System.out.println("\n===== 2. 多级线程切换 =====");

        disposables.add(
                Observable.range(1, 5)
                        .subscribeOn(Schedulers.io()) // 发射在IO线程
                        .doOnNext(i -> System.out.println("发射后: " + i + " - 线程: " + Thread.currentThread().getName()))

                        .observeOn(Schedulers.computation()) // 切换到计算线程
                        .map(i -> {
                            System.out.println("计算线程处理: " + i + " - 线程: " + Thread.currentThread().getName());
                            return i * 10;
                        })

                        .observeOn(Schedulers.single()) // 再切换到单一线程
                        .map(i -> {
                            System.out.println("单一线程处理: " + i + " - 线程: " + Thread.currentThread().getName());
                            return "结果" + i;
                        })

                        .observeOn(Schedulers.io()) // 最后切换到IO线程
                        .subscribe(
                                result -> System.out.println("最终结果: " + result + " - 线程: " + Thread.currentThread().getName()),
                                Throwable::printStackTrace
                        )
        );
    }

    // 3. 背压基础示例
    private static void basicBackpressure() {
        System.out.println("\n===== 3. 背压基础示例 =====");

        // 创建快速发射数据的Flowable
        Flowable<Long> fastProducer = Flowable.interval(10, TimeUnit.MILLISECONDS);

        // 慢消费者
        ResourceSubscriber<Long> slowConsumer = new ResourceSubscriber<Long>() {
            @Override
            protected void onStart() {
                request(1); // 初始请求1个数据
            }

            @Override
            public void onNext(Long item) {
                System.out.println("处理数据: " + item + " - 线程: " + Thread.currentThread().getName());
                try {
                    // 模拟慢处理
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // 处理完一个数据后请求下一个
                request(1);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("背压错误: " + t.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("背压处理完成");
            }
        };

        // 应用背压策略
        disposables.add(
                fastProducer
                        .onBackpressureBuffer( // 设置缓冲区
                                50, // 缓冲区大小
                                () -> System.out.println("缓冲区溢出!"),
                                BackpressureOverflowStrategy.DROP_LATEST
                        )
                        .observeOn(Schedulers.computation(), false, 16) // 设置接收缓冲区
                        .subscribeWith(slowConsumer)
        );
    }

    // 4. 高级背压策略
    private static void advancedBackpressureStrategies() {
        System.out.println("\n===== 4. 高级背压策略 =====");

        // 策略1: BUFFER - 无限缓冲（可能导致OOM）
        Flowable.interval(1, TimeUnit.MILLISECONDS)
                .onBackpressureBuffer() // 默认无限缓冲
                .observeOn(Schedulers.computation())
                .doOnNext(i -> Thread.sleep(10))
                .take(10) // 只取10个避免无限运行
                .subscribe(i -> System.out.println("BUFFER策略: " + i));

        // 策略2: DROP - 丢弃无法处理的数据
        Flowable.interval(1, TimeUnit.MILLISECONDS)
                .onBackpressureDrop(item ->
                        System.out.println("丢弃数据: " + item))
                .observeOn(Schedulers.computation())
                .doOnNext(i -> Thread.sleep(10))
                .take(10)
                .subscribe(i -> System.out.println("DROP策略: " + i));

        // 策略3: LATEST - 只保留最新数据
        Flowable.interval(1, TimeUnit.MILLISECONDS)
                .onBackpressureLatest()
                .observeOn(Schedulers.computation())
                .doOnNext(i -> Thread.sleep(10))
                .take(10)
                .subscribe(i -> System.out.println("LATEST策略: " + i));

        // 策略4: ERROR - 缓冲区溢出时抛出异常
        Flowable.interval(1, TimeUnit.MILLISECONDS)
                .onBackpressureBuffer(16, () -> {
                }, BackpressureOverflowStrategy.ERROR)
                .observeOn(Schedulers.computation())
                .doOnNext(i -> Thread.sleep(10))
                .take(10)
                .subscribe(
                        i -> System.out.println("ERROR策略: " + i),
                        err -> System.err.println("背压错误: " + err)
                );
    }

    // 5. 背压与并行处理
    private static void backpressureWithParallelProcessing() {
        System.out.println("\n===== 5. 背压与并行处理 =====");

        // 模拟大量数据源
        Flowable<Integer> dataSource = Flowable.range(1, 1000);

        // 使用并行处理 + 背压管理
        disposables.add(
                dataSource
                        .parallel(4) // 4个并行流
                        .runOn(Schedulers.computation()) // 每个流在计算线程上运行
                        .map(i -> {
                            // 模拟耗时处理
                            try {
                                Thread.sleep(10);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            return i * 2;
                        })
                        .sequential() // 合并回单个流
                        .onBackpressureBuffer(100) // 设置缓冲区
                        .observeOn(Schedulers.single()) // 在单一线程上观察
                        .subscribe(
                                result -> System.out.println("并行处理结果: " + result),
                                Throwable::printStackTrace,
                                () -> System.out.println("并行处理完成")
                        )
        );
    }
}
